package com.tang.window;

import com.tang.bean.WaterSensor;
import com.tang.functions.WaterSensorMapFunction;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

/**
 * 计数窗口Demo
 *
 * @author tang
 * @since 2023/6/11 15:56
 */
public class CountWindowDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        KeyedStream<WaterSensor, String> keyedStream = env.socketTextStream("192.168.70.141", 7777)
                .map(new WaterSensorMapFunction())
                .keyBy(WaterSensor::getId);

        // 1 滚动窗口
//        WindowedStream<WaterSensor, String, GlobalWindow> window = keyedStream.countWindow(10);
        // 2 滑动窗口
        WindowedStream<WaterSensor, String, GlobalWindow> window = keyedStream.countWindow(10, 5);

        window.process(new ProcessWindowFunction<WaterSensor, String, String, GlobalWindow>() {
            @Override
            public void process(String s, ProcessWindowFunction<WaterSensor, String, String, GlobalWindow>.Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                // 上下文可以拿到window对象，还有其他东西：侧输出流 等等
                long maxTs = context.window().maxTimestamp();
                String maxTime = DateFormatUtils.format(maxTs, "yyyy-MM-dd HH:mm:ss.SSS");
                long count = elements.spliterator().estimateSize();

                out.collect("key=" + s + "的窗口最大时间=" + maxTime + ",包含" + count + "条数据===>" + elements.toString());
            }
        }).print();

        env.execute();
    }

}
